package com.yequan.mq.consumer;

import com.yequan.redis.RedisLock;
import com.yequan.redis.RedisUtil;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * @author yq
 * @date 2020/08/31
 */
public class ConsumerQueue2 {

    private static RedisLock msgRepeat = new RedisLock("msgRepeat");

    public static void main(String[] args) throws InterruptedException, MQClientException {
        /**
         * 1.使用消费者组名称进行实例化（消费者类型：DefaultMQPushConsumer,DefaultMQPullConsumer）
         * 2.同一个GroupName下的多个消费者Consumer可以提高并发处理能力，GroupName需要和消息模式(MessageModel)配合使用
         * 3.RocketMQ支持两种消息模式：
         *      Clusting:一个Consumer只能消费同一个GroupName里面的一部分消息,从而实现负载均衡
         *      Broadcasting:一个Consumer可以消费GroupName里面的所有消息
         */
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name");
        /**
         * 指定nameserver的地址和端口号，可以填写多个使用分号隔开。例如：172.22.41.203；172.21.42.204:9876
         * 达到消除单点故障
         */
        consumer.setNamesrvAddr("127.0.0.1:9876");

        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        /**
         * 消费者订阅消费主题,可以订阅一个或多个,"null"或"*"表示订阅多个主题
         * consumer.subscribe("TopicTest", "tag1||tag2||tagn");
         */
        consumer.subscribe("TopicTest2", "*");
        // Register callback to execute on arrival of messages fetched from brokers.
        consumer.registerMessageListener(new MessageListenerConcurrently() {

            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                msgRepeat.lock();
                try {
                    for (MessageExt msg : msgs) {
                        String key = msg.getKeys();
                        if (RedisUtil.getString(key) != null) {
                            System.out.println("消息已消费，请勿重新消费");
                            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                        }
                        RedisUtil.setString(key, "1");
                        System.out.println(new String(msg.getBody()));
                    }
                } finally {
                    msgRepeat.unlock();
                }
                //System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //启动消费者
        consumer.start();
        System.out.printf("Consumer 2 Started.%n");
    }

}
